package com.atguigu.flink.datastreamapi.transform;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * Created by Smexy on 2022/12/13
 *
 *  Connect连接，用于处理不同类型的流，进行混合。
 *
 *       猪---->猪肉(90%)                    -------->  猪肉+ 牛肉精
 *                          -----> 肉馅机器                            -----> 输出 混合馅料  牛肉馅
 *       牛---->牛肉(10%)                    -------->   牛肉 + 香精
 *
 *   -------------------------------------
 *
 *      泾渭分明
 *
 *          泾河龙王 ---->吐口水 ----->泾河----> 清                            ----> 清
 *
 *                                                   ------>汇入 ---------                 -----> 黄河
 *          渭河龙王---->吐口水 ----->泾河-----> 浑                            -----> 浑
 *
 */
public class Demo6_Connect
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);


        DataStreamSource<Integer> streamSource1 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
        DataStreamSource<String> streamSource4 = env.fromElements("1","2");

       //合并为1个流
        SingleOutputStreamOperator<Long> result = streamSource1.connect(streamSource4)
                                                            .map(new CoMapFunction<Integer, String, Long>()
                                                            {
                                                                int v = 0;
                                                                @Override
                                                                public Long map1(Integer value) throws Exception {
                                                                    v = value;
                                                                    return value + 0l;
                                                                }

                                                                @Override
                                                                public Long map2(String value) throws Exception {
                                                                    System.out.println(v);
                                                                    return Long.parseLong(value);
                                                                }
                                                            });
        result
                     .print();


        env.execute();

    }
}
